package org.apache.rocketmq.client.java.impl.producer;


import apache.rocketmq.v2.Publishing;
import apache.rocketmq.v2.Resource;
import apache.rocketmq.v2.Settings;
import com.google.common.base.MoreObjects;
import com.google.common.util.concurrent.Futures;
import com.google.protobuf.util.Durations;
import org.apache.rocketmq.client.java.impl.ClientSettings;
import org.apache.rocketmq.client.java.impl.ClientType;
import org.apache.rocketmq.client.java.impl.UserAgent;
import org.apache.rocketmq.client.java.retry.ExponentialBackoffRetryPolicy;
import org.apache.rocketmq.client.java.retry.RetryPolicy;
import org.apache.rocketmq.client.java.route.Endpoints;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.Duration;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/10
 * @方法描述：生产者设置类，用于配置RocketMQ生产者客户端的各种参数
 */
public class ProducerSettings extends ClientSettings {

    private static final Logger LOGGER = LoggerFactory.getLogger(ProducerSettings.class);

    //生产者客户端可以发送消息的主题信息的集合
    private final Set<String> topics;

    //生产者客户端可以发送的最大消息体大小
    private volatile int maxBodySizeBytes = 4 * 1024 * 1024;

    //是否验证消息类型的标志
    private volatile boolean validateMessageType = true;

    //构造方法
    public ProducerSettings(String clientId, Endpoints accessPoint, ExponentialBackoffRetryPolicy retryPolicy,
                            Duration requestTimeout, Set<String> topics) {
        super(clientId, ClientType.PRODUCER, accessPoint, retryPolicy, requestTimeout);
        this.topics = topics;
    }


    public int getMaxBodySizeBytes() {
        return maxBodySizeBytes;
    }


    public boolean isValidateMessageType() {
        return validateMessageType;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把ProducerSettings对象转换为ProtocolBuffer格式的Settings对象的方法
     */
    @Override
    public Settings toProtobuf() {
        //创建Publishing对象，这个publishing对象封装了主题信息和是否验证消息类型标志
        final Publishing publishing = Publishing.newBuilder()
                //将主题名称集合转换为Resource对象列表并添加到Publishing中
                .addAllTopics(topics.stream().map(name -> Resource.newBuilder().setName(name).build())
                        .collect(Collectors.toList())).setValidateMessageType(validateMessageType).build();
        //创建Settings.Builder对象，用于构建Settings实例
        final Settings.Builder builder = Settings.newBuilder()
                //设置客户端可以访问的服务端的网络地址信息
                .setAccessPoint(accessPoint.toProtobuf())
                //设置生产者客户端类型
                .setClientType(clientType.toProtobuf())
                //设置请求超时时间
                .setRequestTimeout(Durations.fromNanos(requestTimeout.toNanos()))
                //设置和生产者客户端发布信息相关的配置
                .setPublishing(publishing);
        //设置请求失败重试策略
        return builder.setBackoffPolicy(retryPolicy.toProtobuf()).setUserAgent(UserAgent.INSTANCE.toProtoBuf()).build();
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：生产者客户端应用从服务端发送过来的客户端配置信息的方法
     */
    @Override
    public void applySettingsCommand(Settings settings) {
        //从设置对象中获取发布订阅类型
        final Settings.PubSubCase pubSubCase = settings.getPubSubCase();
        //判断当前接收到的设置类型是不是PUBLISHING类型，因为是生产者客户端，所以应该接收PUBLISHING类型的设置
        if (!Settings.PubSubCase.PUBLISHING.equals(pubSubCase)) {
            //类型不匹配则记录错误日志
            LOGGER.error("[Bug] Issued settings not match with the client type, clientId={}, pubSubCase={}, "
                    + "clientType={}", clientId, pubSubCase, clientType);
            return;
        }
        //以下代码逻辑和toProtobuf()方法中的反过来了，但是数据都一致，无非就是一个是把配置信息转换成Protobuf格式的Settings对象
        //而这个是把信息从Settings中获取，然后设置到当前客户端中，大家自己对比着看一下就行
        final apache.rocketmq.v2.RetryPolicy backoffPolicy = settings.getBackoffPolicy();
        final Publishing publishing = settings.getPublishing();
        RetryPolicy exist = retryPolicy;
        //在这里更新重试策略，这个方法会返回一个更新后的重试策略对象
        this.retryPolicy = exist.updateBackoff(backoffPolicy);
        this.validateMessageType = settings.getPublishing().getValidateMessageType();
        this.maxBodySizeBytes = publishing.getMaxBodySize();
        //这里有一个非常重要的操作，那就是把arrivedFuture这个future对象设置成完成状态，这里一旦设置成完成状态
        //那么调用了ClientImpl对象的awaitSettingSynchronized()方法的线程就可以继续向下执行了
        this.arrivedFuture.setFuture(Futures.immediateVoidFuture());
    }

    @Override
    public String toString() {
        return MoreObjects.toStringHelper(this)
                .add("clientId", clientId)
                .add("clientType", clientType)
                .add("accessPoint", accessPoint)
                .add("retryPolicy", retryPolicy)
                .add("requestTimeout", requestTimeout)
                .add("topics", topics)
                .add("maxBodySizeBytes", maxBodySizeBytes)
                .toString();
    }
}
